package cn.colins.raft.rpc.future.impl;

import cn.colins.raft.rpc.RpcRemoteContext;
import cn.colins.raft.rpc.domin.RpcRemoteMsg;
import cn.colins.raft.rpc.future.RpcWriteFuture;
import cn.hutool.core.util.ReflectUtil;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @Author colins
 * @Description 同步消息
 * @Date 2023/7/15 11:35
 * @Param
 * @return
 **/
public class SyncRpcWriteFuture<R, T> implements RpcWriteFuture<R, T> {

    private static final Logger log = LoggerFactory.getLogger(SyncRpcWriteFuture.class);


    private CountDownLatch latch;
    /**
     * 开始时间戳
     */
    private final long beginTime = System.currentTimeMillis();
    /**
     * 响应
     */
    private R easyRpcResponse;
    /**
     * 超时时间
     */
    private long timeout;
    /**
     * 是否需要超时判断
     */
    private boolean isTimeout;
    /**
     * 请求唯一标识
     */
    private final String requestId;


    public SyncRpcWriteFuture(String requestId,long timeout) {
        this.requestId = requestId;
        this.timeout = timeout;
        this.isTimeout = timeout <= 0L ? false : true;
        this.latch = new CountDownLatch(1);
    }


    @Override
    public <R> R write(Channel channel, T easyRpcRequest) throws InterruptedException {
        RpcRemoteContext.addRequestCache(requestId,this);
        ReflectUtil.setFieldValue(easyRpcRequest,"requestId",requestId);
        channel.writeAndFlush(new RpcRemoteMsg<T>(easyRpcRequest));
        return (R) (isTimeout ? this.get(timeout, TimeUnit.MILLISECONDS) : this.get());
    }

    @Override
    public void setResponse(R response) {
        this.easyRpcResponse = response;
        latch.countDown();
    }

    @Override
    public R getResponse() {
        return easyRpcResponse;
    }

    @Override
    public boolean isTimeout() {
        // true 超时  false 不超时
        if (!isTimeout) {
            return false;
        }

        return System.currentTimeMillis() - beginTime > timeout;
    }

    @Override
    public long getTimeout() {
        return this.timeout;
    }


    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return true;
    }

    @Override
    public boolean isCancelled() {
        return false;
    }

    @Override
    public boolean isDone() {
        return false;
    }

    @Override
    public R get() throws InterruptedException{
        // 获取的时候就需要等待阻塞 等待响应被唤醒
        latch.await();
        return easyRpcResponse;
    }


    @Override
    public R get(long timeout, TimeUnit unit) throws InterruptedException {
        // 获取的时候就需要等待阻塞 等待响应被唤醒
        // 等待一定时长还没响应 代表超时
        if (latch.await(timeout, unit)) {
            return easyRpcResponse;
        }
        log.debug("请求超时了");
        return null;
    }
}
